package com.batchengine.worker;

import com.batchengine.BatchEngineException;
import com.batchengine.Definitions;
import com.batchengine.task.Task;
import com.batchengine.task.TaskContext;
import static com.batchengine.util.LoggingFactory.createLogger;
import static com.batchengine.worker.WorkerType.CONSUMER;
import org.apache.log4j.Logger;

import java.util.concurrent.BlockingQueue;

/**
 * Created by sunil.incubator@yahoo.com.
 * User: sunil.kalva
 * Date: Jun 22, 2009
 */
public class DataConsumer<T> extends Worker {

    private static final Logger logger = createLogger();

    protected BlockingQueue<T> workQ;
    private DataGatherer dataGatherer;

    protected DataConsumer(String name, long taskId, BlockingQueue<T> workQ) {
        super(name, taskId);
        this.workQ = workQ;
    }

    protected DataConsumer(String name, long taskId, ThreadGroup workGroup, BlockingQueue<T> workQ) {
        super(name, taskId, workGroup);
        this.workQ = workQ;
    }

    protected DataConsumer(String name, long taskId, ThreadGroup workGroup, int workerPriority, BlockingQueue<T> workQ) {
        super(name, taskId, workGroup, workerPriority);
        this.workQ = workQ;
    }

    public DataGatherer getDataGatherer() {
        return dataGatherer;
    }

    public void setDataGatherer(DataGatherer dataGatherer) {
        this.dataGatherer = dataGatherer;
    }

    public final WorkerType getType() {
        return CONSUMER;
    }

    public void doWork(TaskContext context) throws InterruptedException {

        try {

            Task task = (Task) context.getContextParam(Definitions.ContextParams.TASK_INSTANCE);
            context.addcontextParam(Definitions.ContextParams.ARGUMENTS, workQ.poll());
            context.addcontextParam(Definitions.ContextParams.OWNER, this);
            task.execute(context);
        } catch (Exception e) {
            throw new BatchEngineException("error#DataConsumer@doWork", e);
        }
    }

    @Override
    public String toString() {
        return "DataConsumer{" +
                ", dataGatherer=" + dataGatherer +
                '}';
    }
}
